package cn.finder.mq.router;

import java.io.UnsupportedEncodingException;

import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;

import cn.finder.mq.FinderMqttClient;


/***
 * 
 * msgone contract:
 * we supply a contract for topic format
 * contract follow:
 * 		publish: msgone/{topic}/{to-client-id}/{client-id}
 * 	   subscribe:msgone/{topic}/{client-id}/+
 * 
 * {to-client-id} is receiver client id .
 * {client-id} is local or sender
 * {topic} is business topic path. example:  star/goods/list
 * 
 * 
 * msgall contract:
 * 	msgall/#
 * 
 * 
 * msgg contract: 
 *    msgg is  group message.
 *    
 * we supply a contract for topic format
 * 	contract follow:
 * 		publish:  msgg/{topic}/{group-id}/{client-id}      
 * 		subscribe:  msgg/{topic}/{group-id}/+    for client
 * 
 * 
 * 
 * 
 * @author finder
 *
 */
public class DefaultTopicRouterMqttClient extends FinderMqttClient{

	public static String TOPIC_PREFIX_MSG_ONE="msgone";
	public static String TOPIC_PREFIX_MSG_ALL="msgall";
	public static String TOPIC_PREFIX_MSG_GROUP="msgg";
	
	public static String MSGONE_PUBLISH_TEMPLATE_FORMAT="msgone/{topic}/{to-client-id}/{client-id}";
	public static String MSGONE_SUBSCRIBE_TEMPLATE_FORMAT="msgone/{topic}/{client-id}/+";
	
	


	public DefaultTopicRouterMqttClient(String clientId, String serverURI) throws MqttException {
		super(clientId, serverURI);
		
	}
	

	/***
	 * subscribeMsgOne 
	 * @param from_clientid
	 * @param to_clientid
	 * @param topic_suffix
	 * @param qos
	 * @throws MqttException 
	 */
	public void subscribeMsgOne(String topic,int qos) throws MqttException{
		String full_topic=String.format("msgone/%s/%s/+", topic,getClientId());
		subscribe(full_topic, qos);
	}
	
	
	
	public void subscribeMsgAll(int qos) throws MqttException{
		String topic=String.format("%s/#", TOPIC_PREFIX_MSG_ALL);
		subscribe(topic, qos);
	}
	
	
	/***
	 *  subscribe
	 * @param topic_prefix
	 * @param from_clientid
	 * @param topic_suffix
	 * @param qos
	 * @throws MqttException
	 */
	public void subscribeMsgOne(String[] topic,int[] qos) throws MqttException{
		
		if(topic.length==qos.length){
			
			int len=qos.length;
			
			String[] topicFilters=new String[len];
			
			for(int i=0;i<len;i++){
				
				topicFilters[i]= String.format("msgone/%s/%s/+", topic[i],getClientId());
				
			}
			
			subscribe(topicFilters,qos);
		}else{
			throw new MqttException(new RuntimeException("parameter length is incorrect"));
		}
		
		
	}
	
	
	/***
	 * msgone/{topic}/{to-client-id}/{client-id}
	 * @param to_clientid
	 * @param topic_suffix
	 * @param msg
	 * @throws MqttException 
	 * @throws MqttPersistenceException 
	 */
	public void publishMsgOne(String topic,String msg,String to_client_id) throws MqttPersistenceException, MqttException{
		
		MqttMessage message=null;
		try {
			message = new MqttMessage(msg.getBytes("UTF-8"));
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		message.setQos(qos);
		message.setRetained(retained); // 设置发送 保留消息
		String full_topic=String.format("msgone/%s/%s/%s", topic,to_client_id,getClientId());
		publish(full_topic, message);
		
		
		
		
	}
	
	
	
	
	
	
	
	/*private String createSubTopic(String topic_prefix,String from_clientid,String topic_suffix){
		
		String topic="";
		if(TOPIC_PREFIX_MSG_ONE.equals(topic_prefix)){
			topic= String.format("%s/%s/%s/%s", TOPIC_PREFIX_MSG_ONE,from_clientid,getClientId(),topic_suffix);
		}else if(TOPIC_PREFIX_MSG_ALL.equals(topic_prefix)){
		
			topic= String.format("%s/%s/%s", TOPIC_PREFIX_MSG_ALL,from_clientid,topic_suffix);
		}
		else if(TOPIC_PREFIX_MSG_GROUP.equals(topic_prefix)){
			topic= String.format("%s/%s/%s", TOPIC_PREFIX_MSG_GROUP,from_clientid,topic_suffix);
		}else{
			topic="";
		}
		
		return topic;
	}*/
	
	
	
	
	

}
